package com.navi.loaddata.loadService.impl.shtWipLoad;


import com.navi.loaddata.core.etlException.ETLException;
import com.navi.loaddata.core.gpEntity.WppShtOpeConf;
import com.navi.loaddata.core.mysqlEntity.EtlConfD;
import com.navi.loaddata.core.repository.impl.GreenPlumRepositoryImpl;

import com.navi.loaddata.core.util.LoggerUtil;

import org.hibernate.SQLQuery;
import org.hibernate.Session;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.*;


/**
 * @author chen
 */
@Service
@Transactional(transactionManager = "secondaryPlatformTransactionManager",
        rollbackFor = {RuntimeException.class, Exception.class})
public class LoadShtWipTmpService {
    private static final Logger logger = LoggerFactory.getLogger(LoadShtWipTmpService.class);
    private LoggerUtil loggerUtil = new LoggerUtil(this.getClass());

    public static final List<String> ASHT_QUERY_COLUMN =Arrays.asList("panel_id","sn","lot_id","bare_glass_id",
            "bare_product","product_id_erp","workorder_id","workorder_id_rtp","thickness","glass_judge_ary","glass_grade_ary","tft_pnl_grade",
            "glass_judge_cf","glass_grade_cf","cf_pnl_grade","oc_pnl_grade_ct1","oc_pnl_grade_c_insp","grade_wns_ship","oc_pnl_losscode_ct1","oc_pnl_losscode_c_insp",
            "pol_lot_id_tft","pol_part_no_tft","pol_lot_id_cf","pol_part_no_cf","productiontype","ownertype","ppbox_id","pallet_id",
            "dest_customer_id","tft_start_timestamp","tft_shipping_timestamp","cf_start_timestamp","cf_shipping_timestamp","oc_start_timestamp","oc1_bank_in_timestamp","oc2_start_timestamp",
            "oc_shipping_timestamp","rmatimestamp","in_wms_po_id","salesorder_id");
    public static final List<String> FSHT_QUERY_COLUMN =Arrays.asList("panel_id","sn","lot_id","bare_glass_id",
            "bare_product","product_id_erp","workorder_id","workorder_id_rtp","thickness","glass_judge_ary","glass_grade_ary","tft_pnl_grade",
            "glass_judge_cf","glass_grade_cf","cf_pnl_grade","oc_pnl_grade_ct1","oc_pnl_grade_c_insp","grade_wns_ship","oc_pnl_losscode_ct1","oc_pnl_losscode_c_insp",
            "pol_lot_id_tft","pol_part_no_tft","pol_lot_id_cf","pol_part_no_cf","productiontype","ownertype","ppbox_id","pallet_id",
            "dest_customer_id","tft_start_timestamp","tft_shipping_timestamp","cf_start_timestamp","cf_shipping_timestamp","oc_start_timestamp","oc1_bank_in_timestamp","oc2_start_timestamp",
            "oc_shipping_timestamp","rmatimestamp","in_wms_po_id","salesorder_id");
    public static final List<String> CSHT_QUERY_COLUMN =Arrays.asList("panel_id","tft_glass_id","cf_glass_id","sn","lot_id","bare_glass_id",
            "bare_product","product_id_erp","workorder_id","workorder_id_rtp","thickness","glass_judge_ary","glass_grade_ary","tft_pnl_grade",
            "glass_judge_cf","glass_grade_cf","cf_pnl_grade","oc_pnl_grade_ct1","oc_pnl_grade_c_insp","grade_wns_ship","oc_pnl_losscode_ct1","oc_pnl_losscode_c_insp",
            "pol_lot_id_tft","pol_part_no_tft","pol_lot_id_cf","pol_part_no_cf","productiontype","ownertype","ppbox_id","pallet_id",
            "dest_customer_id","tft_start_timestamp","tft_shipping_timestamp","cf_start_timestamp","cf_shipping_timestamp","oc_start_timestamp","oc1_bank_in_timestamp","oc2_start_timestamp",
            "oc_shipping_timestamp","rmatimestamp","in_wms_po_id","salesorder_id");


    @Autowired
    private GreenPlumRepositoryImpl gpDao;

    public void loadMain1(EtlConfD etlConfD) throws ETLException {
        Session session = gpDao.getEntityManagerSession();
        try {
            logger.info("Load data  To Tmp start");
            logger.info("etlConfD message : etlconfd startTime: "+etlConfD.getRunStartTimestamp()+
                    "---etlconfd endTime:  "+etlConfD.getRunEndTimestamp() +"---etlconfd ext1"+etlConfD.getExt1());
            loadDataToInfoTmp(session, etlConfD);   // hisAndExtToTmp
//            loadDataToOpeInfoTmp(session, etlConfD);
            logger.info("Load data  To Tmp finish");
        } catch (ETLException e) {
            logger.error(LoggerUtil.stackTraceToString(e));
            throw new ETLException(e.getMessage());
        }
    }

    public void loadMain2(EtlConfD etlConfD) throws ETLException {
        Session session = gpDao.getEntityManagerSession();
        try {
            logger.info("Load data  To TmpSum start");
            loadDataTmpToInfo(session, etlConfD);
//            loadDatOpeTmpToOpeInfo(session, etlConfD);
            logger.info("Load data  To TmpSum finish");
            logger.info("etlConfD message : etlconfd startTime: "+etlConfD.getRunStartTimestamp()+
                    "---etlconfd endTime:  "+etlConfD.getRunEndTimestamp() +"---etlconfd ext1"+etlConfD.getExt1());
        } catch (ETLException e) {
            logger.error(LoggerUtil.stackTraceToString(e));
            throw new ETLException(e.getMessage());
        }
    }

    /**
     * Load MES HIS
     *
     * @param session
     * @param etlConfD
     */

    private void loadDataToInfoTmp(Session session, EtlConfD etlConfD) throws ETLException {
        String tableA = "wpp_asht_ope_his";
        String tableC = "wpp_csht_ope_his";
        String tableF = "wpp_fsht_ope_his";
        hisAndExtToTmp(session, etlConfD, tableA);
        hisAndExtToTmp(session, etlConfD, tableC);
        hisAndExtToTmp(session, etlConfD, tableF);

    }


    private void hisAndExtToTmp(Session session, EtlConfD etlConfD, String tableName) throws ETLException {
        int countSize = 0;
       /* StringBuffer sqlbuffColumn = new StringBuffer();
        sqlbuffColumn.append(" SELECT COLUMN_NAME FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_NAME = 'wpp_sht_info'  ");
        sqlbuffColumn.append(" and COLUMN_NAME  in( ");
        sqlbuffColumn.append(" SELECT COLUMN_NAME FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_NAME = '" + tableName + "' ");
        sqlbuffColumn.append(" UNION  ");
        sqlbuffColumn.append(" SELECT COLUMN_NAME FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_NAME = 'wpp_sht_ope_his_ext'  )");

        sqlbuffColumn.append(" and COLUMN_NAME not in( ");
        sqlbuffColumn.append(" SELECT a.COLUMN_NAME  FROM ");
        sqlbuffColumn.append("(SELECT COLUMN_NAME FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_NAME = '" + tableName + "' )a,  ");
        sqlbuffColumn.append(" (SELECT COLUMN_NAME FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_NAME = 'wpp_sht_ope_his_ext')c  ");
        sqlbuffColumn.append(" where c.COLUMN_NAME =a.COLUMN_NAME) ");
        sqlbuffColumn.append(" ORDER BY ORDINAL_POSITION  ");*/
        List<String> tableColumns = new ArrayList<>();
        try {
       if(tableName.equals("wpp_asht_ope_his")){
           tableColumns =ASHT_QUERY_COLUMN;
       }else if(tableName.equals("wpp_fsht_ope_his")){
           tableColumns =FSHT_QUERY_COLUMN;
       }else if(tableName.equals("wpp_csht_ope_his")){
           tableColumns =CSHT_QUERY_COLUMN;
       }


          //  tableColumns = gpDao.findBySQL(sqlbuffColumn.toString());  //获取表中所有列名
        } catch (Exception e) {
            logger.error(LoggerUtil.stackTraceToString(e));
            throw new ETLException("ERROR hisAndExtToTmp GET COLUMN_NAME:" + e.getMessage());
        }
        StringBuffer sqlbuffTmp = new StringBuffer();
        sqlbuffTmp.append(" INSERT INTO STAGE.wpp_sht_info_tmp (panel_id ");
        tableColumns.forEach(column -> {
            if (column.equals("panel_id") || column.equals("mes_timestamp") ||
                    column.equals("etl_timestamp") || column.equals("db_timestamp")
                    || column.equals("lot_id")|| column.equals("tft_glass_id")|| column.equals("cf_glass_id")
                    || column.equals("oc_shipping_timestamp")
                    ) return;

            sqlbuffTmp.append("," + column);
        });
        sqlbuffTmp.append(",owner_Id");
        if(tableName.contains("asht")){
            sqlbuffTmp.append(",PRODUCT_ID_ARY");
            sqlbuffTmp.append(",NG_MARK_ARY");
            sqlbuffTmp.append(",lot_id");


        }
        if(tableName.contains("fsht")){
            sqlbuffTmp.append(",PRODUCT_ID_CF");
            sqlbuffTmp.append(",NG_MARK_CF");
            sqlbuffTmp.append(",group_id");

        }
         if(tableName.contains("csht")){
             sqlbuffTmp.append(",NG_MARK_OC");
             sqlbuffTmp.append(",PRODUCT_ID_OC1");
             sqlbuffTmp.append(",PRODUCT_ID_OC2");
             sqlbuffTmp.append(",PRODUCT_ID_BONGING");
             sqlbuffTmp.append(",group_id");
             sqlbuffTmp.append(",oc_shipping_timestamp");
         }
        sqlbuffTmp.append(",tft_glass_id,cf_glass_id ");
        sqlbuffTmp.append(" ) ");
        //将ope_his 与his_ext 左连接,转为info表结构  一个panel_id只对应一笔记录
        sqlbuffTmp.append(" select F.glass_id as panel_id");
        tableColumns.forEach(column -> {
            if (column.equals("panel_id") || column.equals("mes_timestamp") ||
                    column.equals("etl_timestamp") || column.equals("db_timestamp")
                    ||column.equals("lot_id") ||column.equals("tft_glass_id")
                    ||column.equals("cf_glass_id") || column.equals("oc_shipping_timestamp")
                    ) return;
            sqlbuffTmp.append(",F." + column);
        });
        sqlbuffTmp.append(",F.owner as owner_Id");
        if(tableName.contains("asht")){
            sqlbuffTmp.append(", F.prod_id  as PRODUCT_ID_ARY");
            sqlbuffTmp.append(",case when F.ope_id='A9960' then F.NG_MARK else null END as NG_MARK_ARY");
            sqlbuffTmp.append(",F.lot_id");
            //sqlbuffTmp.append(",tft_glass  as tft_glass_id ");
            sqlbuffTmp.append(",F.glass_id  as tft_glass_id ");
            sqlbuffTmp.append(",cf_glass  as cf_glass_id ");
        }
        if(tableName.contains("fsht")){
            sqlbuffTmp.append(",F.prod_id  as PRODUCT_ID_CF");
            sqlbuffTmp.append(",case when F.ope_id='C9960'   then F.NG_MARK else null END as NG_MARK_CF  ");
            sqlbuffTmp.append(",F.lot_id as group_id");
            sqlbuffTmp.append(",tft_glass  as tft_glass_id ");
           // sqlbuffTmp.append(",cf_glass  as cf_glass_id ");
            sqlbuffTmp.append(",F.glass_id   as cf_glass_id ");
        }
        if(tableName.contains("csht")) {

            sqlbuffTmp.append(",case when F.ope_id='L9100'  then F.NG_MARK else null  END  as NG_MARK_OC  ");
            sqlbuffTmp.append(",case when  F.ope_id='L1100' then  F.prod_id else null END  as PRODUCT_ID_OC1");
            sqlbuffTmp.append(",case when  F.ope_id='L5100' then  F.prod_id else null END  as  PRODUCT_ID_OC2");
            sqlbuffTmp.append(",case when  F.ope_id='L7200' then  F.prod_id else null END  as  PRODUCT_ID_BONGING");
            sqlbuffTmp.append(",F.lot_id as group_id");
            sqlbuffTmp.append(",case when  F.ope_id='WMS' then  F.logof_timestamp else null END  as  OC_SHIPPING_TIMESTAMP");
            sqlbuffTmp.append(",F.tft_glass_id");
            sqlbuffTmp.append(",F.cf_glass_id");
        }



       // sqlbuffTmp.append(" from  (select H.*,E.*,AG.cf_glass,FG.tft_glass from (select A.*  from ( ");
       // sqlbuffTmp.append(" select O.*, row_number() over (partition by O.glass_id order by O.mes_timestamp) as group_idx  ");
        sqlbuffTmp.append(" from  (select H.*,E.*,AG.cf_glass,FG.tft_glass from (");
        sqlbuffTmp.append(" select O.*  ");
        sqlbuffTmp.append("  from " + tableName + " O   where");
        sqlbuffTmp.append("   O.DB_TIMESTAMP BETWEEN '" + etlConfD.getRunStartTimestamp() + "' AND '" + etlConfD.getRunEndTimestamp() + "'");
       // sqlbuffTmp.append(" )A where A.group_idx = 1 order BY A.glass_id )H LEFT JOIN (SELECT S.*  FROM WPP_SHT_OPE_HIS_EXT  S  ");
        sqlbuffTmp.append(" )H LEFT JOIN (SELECT S.*  FROM WPP_SHT_OPE_HIS_EXT  S  ");
       // sqlbuffTmp.append(" )A order BY A.glass_id )H LEFT JOIN (SELECT S.*  FROM WPP_SHT_OPE_HIS_EXT  S  ");
        sqlbuffTmp.append(" ORDER BY S.PANEL_ID)E On H.glass_id=E.PANEL_ID   ");
        sqlbuffTmp.append(" LEFT JOIN (SELECT DISTINCT(A1.cf_glass_id) AS cf_glass,A1.tft_glass_id AS tft_glass FROM " +
                " WPP_CSHT_OPE_HIS A1 )AG ON H.glass_id = AG.tft_glass  and AG.cf_glass is not NULL ");
        sqlbuffTmp.append(" LEFT JOIN (SELECT DISTINCT(F1.tft_glass_id) AS tft_glass,F1.cf_glass_id AS cf_glass FROM " +
                " WPP_CSHT_OPE_HIS F1)FG ON H.glass_id = FG.cf_glass  and FG.tft_glass is not NULL");
        sqlbuffTmp.append("  )F  where 1=1 ");
        try {
          //  System.out.println(sqlbuffTmp);
           // logger.info("from "+ tableName +"   to tmp SQL:----------"+sqlbuffTmp);
            SQLQuery queryLoadStep = session.createSQLQuery(sqlbuffTmp.toString());
            int count = queryLoadStep.executeUpdate();
            logger.info(tableName+"  hisAndExtToTmp count :" +count);
        } catch (Exception e) {
            logger.error(LoggerUtil.stackTraceToString(e));
            logger.error("************   sqlbuffTmp.toString():  **********    "+sqlbuffTmp.toString());
            e.printStackTrace();
            throw new ETLException("ERROR hisAndExtToTmp INSERT:" + e.getMessage());

        }
    }

    //tmp to info

    public int loadDataTmpToInfo(Session session, EtlConfD etlConfD) throws ETLException {
        int countSize = 0;
        try {
            empotyTmp(session, "stage.wpp_sht_info_merge");
            empotyTmp(session, "stage.wpp_sht_info_distinct");
            String tableName = "wpp_sht_info";
            StringBuffer sqlbuffColumn = new StringBuffer();
            long startTimes = System.currentTimeMillis();
            sqlbuffColumn.append(" SELECT COLUMN_NAME FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_NAME = '" + tableName + "' " +
                    " ORDER BY ORDINAL_POSITION;");
            List<String> tableColumns = gpDao.findBySQL(sqlbuffColumn.toString());  //获取表中所有列名
            long endTimes = System.currentTimeMillis();
            long minutes = (endTimes - startTimes) / 60;
            logger.info("Job {}--- loadDataTmpToInfo  获取 {} 表结构执行了：{} 秒 ", "LoadShtWipTmpService" ,tableName,minutes);
            tmpToDistinct(session, tableColumns,"wpp_sht_info_tmp","wpp_sht_info_distinct");
            distinctToMerge(session, tableColumns,"wpp_sht_info_distinct","wpp_sht_info_merge","wpp_sht_info");
            mergeToInfo(session, tableColumns,"wpp_sht_info_merge","wpp_sht_info");

            List<String> distinct = gpDao.findBySQL(" SELECT d.panel_id FROM stage.wpp_sht_info_distinct d where 1=1");
            List<String> merges = gpDao.findBySQL(" SELECT m.panel_id FROM stage.wpp_sht_info_merge m where 1=1");

             empotyTmp(session, "stage.wpp_sht_info_merge");
            empotyTmp(session, "stage.wpp_sht_info_distinct");
            updateWppShtInfoData(session);
            logger.info("TRUNCATE :merges.size()=" + merges.size() + "---distinct.size()=" + distinct.size());
            if (merges.size() > 0 && distinct.size() > 0) {
                countSize = distinct.size();
                empotyTmp(session, "stage.wpp_sht_info_tmp");
                logger.info(" 清空 distinct，merge，tmp");
            } else {
                logger.info(" 只清空 distinct，merge，不清空infotmp");
            }

        } catch (Exception e) {
            logger.error(LoggerUtil.stackTraceToString(e));
            if ("ERROR".equals(e.getMessage().substring(0, 5))) {
                throw e;
            }
            throw new ETLException("ERROR hisAndExt_SUM INSERT:" + e.getMessage());
        }
        return countSize;


    }

    private void tmpToDistinct(Session session, List<String> tableColumns,String tmpTableName,String distinctTableName) throws ETLException {
        //tmpTableName ="wpp_sht_info_tmp";
        //distinctTableName="wpp_sht_info_distinct";
        StringBuffer sqlbuffDistinct = new StringBuffer();
        try {
            //存储中间结果
            sqlbuffDistinct.append(" INSERT INTO STAGE."+distinctTableName+" (panel_id  ");
            tableColumns.forEach(column -> {
                if (column.equals("panel_id")) return;
                sqlbuffDistinct.append("," + column);
            });
            sqlbuffDistinct.append(" ) ");

            sqlbuffDistinct.append(" SELECT DISTINCT ON (PANEL_ID) ");
            sqlbuffDistinct.append("  PANEL_ID");
            //对每一字段做更新处理，若为空，保留原值
            tableColumns.forEach(column -> {
                if (column.equals("panel_id")) return;
                sqlbuffDistinct.append(", first_value(" + column + " ) over (partition by panel_id order by ");
//            sqlbuffDistinct.append( " (CASE WHEN "+column+" IS NULL THEN NULL ELSE db_timestamp END) DESC NULLS LAST) AS "+column  );
//                sqlbuffDistinct.append("db_timestamp  DESC ) AS " + column);
                sqlbuffDistinct.append( " (CASE WHEN "+column+" IS NULL THEN NULL ELSE db_timestamp END)," );
                sqlbuffDistinct.append( column+" is not null DESC ) AS "+column  );

            });
            sqlbuffDistinct.append("  FROM stage."+tmpTableName+"  where 1=1");

            SQLQuery queryLoadStep = session.createSQLQuery(sqlbuffDistinct.toString());
          int count =   queryLoadStep.executeUpdate();
            logger.info(tmpTableName+"  To  "+distinctTableName+"  count :" +count);
        } catch (Exception e) {
            logger.error("************   sqlbuffDistinct.toString():  **********    "+sqlbuffDistinct.toString());
            logger.error(LoggerUtil.stackTraceToString(e));
            throw new ETLException("ERROR tmpToDistinct INSERT:" + e.getMessage());
        }

    }


    private void distinctToMerge(Session session, List<String> tableColumns,String distinctTableName,String mergeTableNmae,String finalTableName) throws ETLException {
        //distinctTableName="wpp_sht_info_distinct";
        //mergeTableNmae="wpp_sht_info_merge";
        //finalTableName="wpp_sht_info";
        //合并写入
        StringBuffer sqlbuffMerge = new StringBuffer();
        try {

            sqlbuffMerge.append(" INSERT INTO STAGE."+mergeTableNmae+" (panel_id  ");

            tableColumns.forEach(column -> {
                if (column.equals("panel_id")) return;
                sqlbuffMerge.append("," + column);
            });
            sqlbuffMerge.append(" ) ");
            sqlbuffMerge.append(" SELECT D.PANEL_ID ");
            //对每一字段做更新处理
            tableColumns.forEach(column -> {
                if (column.equals("panel_id")) return;
                sqlbuffMerge.append(", coalesce(D." + column + ", I." + column + ") " + column);
            });
            sqlbuffMerge.append("  FROM STAGE."+distinctTableName+" AS  D  LEFT JOIN  "+finalTableName+"  I  USING (PANEL_ID)");  //left join ? inner join

            SQLQuery queryLoadStep = session.createSQLQuery(sqlbuffMerge.toString());
            int count =   queryLoadStep.executeUpdate();
            logger.info(distinctTableName+"  To  "+mergeTableNmae+"  count :" +count);
        } catch (Exception e) {
            logger.error("************   sqlbuffMerge.toString():  **********    "+sqlbuffMerge.toString());
            logger.error(LoggerUtil.stackTraceToString(e));
            throw new ETLException("ERROR tmpToDistinct INSERT:" + e.getMessage());
        }
    }

    private void mergeToInfo(Session session, List<String> tableColumns,String mergeTableName,String finalTableName) throws ETLException {
       // mergeTableName ="WPP_SHT_INFO";
        //finalTableName ="WPP_SHT_INFO_MERGE";
        String sqlDelete = " DELETE FROM "+finalTableName+"  I USING STAGE."+mergeTableName+" AS M  WHERE I.PANEL_ID=M.PANEL_ID ";
        StringBuffer sqlbuffInfo = new StringBuffer();
        try {
            // DELETE USING，删除全量表的符合条件的记录。
            SQLQuery queryLoadStepD = session.createSQLQuery(sqlDelete);
            queryLoadStepD.executeUpdate();
            //  INSERT wpp_sht_info

            sqlbuffInfo.append(" INSERT INTO "+finalTableName+" (panel_id");
            tableColumns.forEach(column -> {
                if (column.equals("panel_id")) return;
                sqlbuffInfo.append("," + column);
            });
            sqlbuffInfo.append(" ) ");
            //  sqlbuffInfo.append(" SELECT D.* FROM WPP_SHT_INFO_DISTINCT D LEFT JOIN STAGE.WPP_SHT_INFO_MERGE M USING (PANEL_ID) WHERE M.* IS NULL   " );
            //  sqlbuffInfo.append(" UNION ALL   "   );
            sqlbuffInfo.append("  SELECT panel_id");
            tableColumns.forEach(column -> {
                if (column.equals("panel_id")) return;
                sqlbuffInfo.append("," + column);
            });
            sqlbuffInfo.append("  FROM STAGE."+mergeTableName);
            SQLQuery queryLoadStep = session.createSQLQuery(sqlbuffInfo.toString());
            int count =   queryLoadStep.executeUpdate();
            logger.info(mergeTableName+"  To  "+finalTableName+" count :" +count);
        } catch (Exception e) {
            logger.error("************   sqlDelete.toString():  **********    "+sqlDelete.toString());
            logger.error("************   sqlbuffInfo.toString():  **********    "+sqlbuffInfo.toString());
            logger.error(LoggerUtil.stackTraceToString(e));
            throw new ETLException("ERROR mergeToInfo :" + e.getMessage());
        }


    }

    //清空临时表
    private void empotyTmp(Session session, String tableName) throws ETLException {
        StringBuffer sqlbuff = new StringBuffer();
        try {
            sqlbuff.append(" TRUNCATE TABLE " + tableName);
            SQLQuery queryLoadStep = session.createSQLQuery(sqlbuff.toString());
            int count =   queryLoadStep.executeUpdate();
            logger.info("TRUNCATE table :"+tableName+"   count :" +count);
        } catch (Exception e) {
            logger.error("************   empotyTmp.toString():  **********    "+sqlbuff.toString());
            logger.error(LoggerUtil.stackTraceToString(e));
            throw new ETLException("ERROR TRUNCATE TABLE " + tableName+" :" + e.getMessage());
        }
    }


    /**
     * 对wpp_sht_info 表进行大板，小板 array,cf阶段数据互补
     */
    private void updateWppShtInfoData(Session session)throws ETLException{
        StringBuffer sqlbuff = new StringBuffer();
        try {
            //用CF大板补全ARRAY,OC板中CF数据
            sqlbuff.append(" update sor.wpp_sht_info a SET product_id_cf = b.product_id_cf ,");
            sqlbuff.append(" glass_grade_cf =b.glass_grade_cf,glass_judge_cf = b.glass_judge_cf,");
            sqlbuff.append(" cf_pnl_grade =b.cf_pnl_grade,cf_start_timestamp = b.cf_start_timestamp ,");
            sqlbuff.append(" cf_shipping_timestamp=b.cf_shipping_timestamp,ng_mark_cf =b.ng_mark_cf ");
            sqlbuff.append(" from (SELECT * FROM sor.wpp_sht_info i where i.panel_id in");
            sqlbuff.append(" (SELECT glass_id FROM sor.wpp_fsht_ope_his) )b where a.cf_glass_id=b.panel_id; ");
            //用 ARRAY大板数据  补全OC，CF板的ARRAY数据
            sqlbuff.append(" update sor.wpp_sht_info a SET product_id_ary = b.product_id_ary ,glass_grade_ary =b.glass_grade_ary,");
            sqlbuff.append(" glass_judge_ary = b.glass_judge_ary,tft_pnl_grade =b.tft_pnl_grade,tft_start_timestamp = b.tft_start_timestamp ,");
            sqlbuff.append(" tft_shipping_timestamp=b.tft_shipping_timestamp,ng_mark_ary =b.ng_mark_ary");
           // sqlbuff.append(" ,product_id_oc1=b.product_id_oc1,product_id_erp=b.product_id_erp,workorder_id =b.workorder_id");
            // sqlbuff.append(" ,oc_start_timestamp = b.oc_start_timestamp");
            sqlbuff.append(" from (SELECT * FROM sor.wpp_sht_info i where i.panel_id in(SELECT glass_id FROM sor.wpp_asht_ope_his ))b");
            sqlbuff.append(" where a.tft_glass_id=b.panel_id;");
            //用 ARRAY大板数据  补全OC板的OC数据,CF有自己数据，不需要补值
           sqlbuff.append(" update sor.wpp_sht_info a SET ");
           sqlbuff.append(" product_id_oc1=b.product_id_oc1,product_id_erp=b.product_id_erp,workorder_id =b.workorder_id ");
             sqlbuff.append(" ,oc_start_timestamp = b.oc_start_timestamp ");
            sqlbuff.append(" from (SELECT * FROM sor.wpp_sht_info i where i.panel_id in(SELECT glass_id FROM sor.wpp_asht_ope_his ))b");
            sqlbuff.append(" where a.tft_glass_id=b.panel_id and a.panel_id not in(b.tft_glass_id, b.cf_glass_id);");


            long startTimes = System.currentTimeMillis();
            SQLQuery queryLoadStep = session.createSQLQuery(sqlbuff.toString());
            int count =   queryLoadStep.executeUpdate();
            long endTimes = System.currentTimeMillis();
            long minutes = (endTimes - startTimes) / 60;
            logger.info("用CF大板补全ARRAY,OC板中CF数据;用 ARRAY大板数据  补全OC，CF板的ARRAY数据;  minutes:{}s---count{}:", minutes,count);
        } catch (Exception e) {
            logger.error("************  updateWppShtInfoData SQL:  **********    "+sqlbuff.toString());
            logger.error(LoggerUtil.stackTraceToString(e));
            throw new ETLException("ERROR updateWppShtInfoData :" + e.getMessage());
        }
    }


}
